/*! \file    record.h
 * \author   Lorenzo Miniero <lorenzo@meetecho.com>
 * \copyright GNU General Public License v3
 * \brief    Audio/Video processer
 * \details  Implementation of a simple processer utility that plugins
 * can make use of to record audio/video frames to a Janus file. This
 * file just saves RTP frames in a structured way, so that they can be
 * post-processed later on to get a valid container file (e.g., a .opus
 * file for Opus audio or a .webm file for VP8 video) and keep things
 * simpler on the plugin and core side.
 * \note If you want to record both audio and video, you'll have to use
 * two different processers. Any muxing in the same container will have
 * to be done in the post-processing phase.
 * 
 * \ingroup core
 * \ref core
 */
 
#include <arpa/inet.h>
#include <sys/stat.h>
#include <errno.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/wait.h>

#include <libgen.h>

#include <glib.h>
#include <jansson.h>

#include "record.h"
#include "postprocessing.h"
#include "debug.h"
#include "utils.h"

#define JANUS_PROCESSER_NAME "JANUS Processer"
#define JANUS_DEFAULT_SCRIPT "/opt/janus/share/janus/streams/processer.sh"
//#define JANUS_DEBUG_DISABLE

static volatile gint initialized = 0, stopping = 0;
static GThread *processer_thread;
static void *janus_processer(void *data);

static GHashTable *audio_files, *video_files;
static const char *pattern[2] = { "-audio.mjr", "-video.mjr" };
static GHashTable *pattern_hash[2], *tuple_hash[2];
static char processer_script[256];

typedef struct janus_processer_message {
	char prefix_name[256];
	char dir[256];
	char codec[32];
	int type;
	char business_model[256];
	char business_id[256];
} janus_processer_message;

static GAsyncQueue *messages = NULL;
static janus_processer_message exit_message;

static void janus_processer_message_free(janus_processer_message *msg) {
	if(!msg || msg == &exit_message)
		return;

	JANUS_LOG(LOG_INFO, "Free msg %p %s\n", msg, msg->prefix_name);
	g_free(msg);
}

void janus_processer_init(void) {
	JANUS_LOG(LOG_INFO, "Initializing processer code\n");
	audio_files = g_hash_table_new_full(g_str_hash, g_str_equal, NULL, (GDestroyNotify)janus_processer_message_free);
	video_files = g_hash_table_new_full(g_str_hash, g_str_equal, NULL, (GDestroyNotify)janus_processer_message_free);
	messages = g_async_queue_new_full((GDestroyNotify) janus_processer_message_free);
	/* This is the callback we'll need to invoke to contact the server */
	g_atomic_int_set(&initialized, 1);

	pattern_hash[0] = audio_files;
	pattern_hash[1] = video_files;
	tuple_hash[1] = audio_files;
	tuple_hash[0] = video_files;

	if(getenv("JANUS_SCRIPT_PATH"))
		snprintf(processer_script, sizeof(processer_script), "%s", getenv("PATH"));
	else
		snprintf(processer_script, sizeof(processer_script), "%s", JANUS_DEFAULT_SCRIPT);

	/* Launch the thread that will handle incoming messages */
	GError *error = NULL;
	processer_thread = g_thread_try_new("Processer", janus_processer, NULL, &error);
	if(error != NULL) {
		g_atomic_int_set(&initialized, 0);
		JANUS_LOG(LOG_ERR, "Got error %d (%s) trying to launch the processer thread...\n", error->code, error->message ? error->message : "??");
		return;
	}
	JANUS_LOG(LOG_INFO, "%s initialized!\n", JANUS_PROCESSER_NAME);
}

void janus_processer_deinit(void) {
	if(!g_atomic_int_get(&initialized))
		return;
	g_atomic_int_set(&stopping, 1);

	g_async_queue_push(messages, &exit_message);
	if(processer_thread != NULL) {
		g_thread_join(processer_thread);
		processer_thread = NULL;
	}

	/* FIXME We should destroy the sessions cleanly */
	g_hash_table_destroy(audio_files);
	audio_files = NULL;
	g_hash_table_destroy(video_files);
	video_files = NULL;
	g_async_queue_unref(messages);
	messages = NULL;

	g_atomic_int_set(&initialized, 0);
	g_atomic_int_set(&stopping, 0);
	JANUS_LOG(LOG_INFO, "%s destroyed!\n", JANUS_PROCESSER_NAME);
}

void janus_processer_incoming_task(const char* filename, const char* dir, 
	const char* codec, const int type,
	const char* business_model, const char* business_id) {
	char *end;
	if(g_atomic_int_get(&stopping) || !g_atomic_int_get(&initialized)) {
		/* Janus is closing or the plugin is */
		return;
	}

	if( (type != JANUS_RECORDER_VIDEO) &&  
		(type != JANUS_RECORDER_AUDIO) ) {
		return;
	}

	end = strstr(filename, pattern[type]);
	if(!end) {
		return;
	}

	janus_processer_message *proc_msg = g_malloc0(sizeof(janus_processer_message));
	if(proc_msg == NULL)
		return;

	memset(proc_msg, 0, sizeof(*proc_msg));
	memcpy(proc_msg->prefix_name, filename, (end - filename));
	snprintf(proc_msg->dir, sizeof(proc_msg->dir), "%s", dir);
	snprintf(proc_msg->codec, sizeof(proc_msg->codec), "%s", codec);
	proc_msg->type = type;
	snprintf(proc_msg->business_model, sizeof(proc_msg->business_model), "%s", business_model);
	snprintf(proc_msg->business_id, sizeof(proc_msg->business_id), "%s", business_id);
	
	g_async_queue_push(messages, proc_msg);
}

/* put all the parameters into an environment */
static char **fill_envp(const char* dir, const char* filename, 
	const char* audio_code,  const char* video_codec,
	const char* business_model, const char* business_id)
{
	int i = 0, rc;
	char **envp = calloc(sizeof(char *), 8);

	rc = asprintf(&envp[i++], "dir=%s", dir);
	rc = asprintf(&envp[i++], "filename=%s", filename);
	rc = asprintf(&envp[i++], "video_codec=%s", video_codec);
	rc = asprintf(&envp[i++], "audio_codec=%s", audio_code);
	rc = asprintf(&envp[i++], "business_model=%s", business_model);
	rc = asprintf(&envp[i++], "business_id=%s", business_id);
	rc = asprintf(&envp[i++], "%s=%s", "PATH",
		getenv("PATH") ? : "/bin:/usr/bin:/sbin:/usr/sbin");
	rc = asprintf(&envp[i++], "%s=%s", "HOME", getenv("HOME") ? : "/");

	return envp;
}

/* Call a script with a par file and env vars */
static void run_script(const char* dir, const char* filename, 
	const char* audio_code, const char* video_codec,
	const char* business_model, const char* business_id)
{
	int pid;
	char **envp, **curr;

	JANUS_LOG(LOG_INFO, "forking and execle'ing %s", processer_script);
	envp = fill_envp(dir, filename, audio_code, video_codec, business_model, business_id);
	/* call script */
	pid = fork();
	if (pid) {
		waitpid(pid, NULL, 0);
		for (curr = envp; *curr; curr++) free(*curr);
		free(envp);
		return;
	} else if (pid == 0) {
		/* close fd's? */

#ifdef JANUS_DEBUG_DISABLE
		close(0);
		close(1);
		close(2);
#endif
		/* exec script */
		execle(processer_script, processer_script,
		       "janus_processer", NULL, envp);
		JANUS_LOG(LOG_ERR, "script %s failed: %m", processer_script);
		exit(1);
	}
}

/* Call a script with a par file and env vars */
static void run_script_async(const char* dir, const char* filename, 
		const char* audio_code, const char* video_codec,
		const char* business_model, const char* business_id)
{
	int pid;
	char **envp, **curr;

	JANUS_LOG(LOG_INFO, "forking and execle'ing %s\n", processer_script);
	envp = fill_envp(dir, filename, audio_code, video_codec, business_model, business_id);
	/* call script */
	pid = fork();
	if (pid) {
		for (curr = envp; *curr; curr++) free(*curr);
		free(envp);
		return;
	} else if (pid == 0) {
		/* close fd's? */

#ifdef JANUS_DEBUG_DISABLE
		close(0);
		close(1);
		close(2);
#endif
		setsid();
		/* exec script */
		execle(processer_script, processer_script,
		       "janus_processer", NULL, envp);
		JANUS_LOG(LOG_ERR, "script %s failed: %m", processer_script);
		exit(1);
	}
}

static void *janus_processer(void *data) {
	JANUS_LOG(LOG_VERB, "Joining Processer thread\n");
	janus_processer_message *msg = NULL;

	while(g_atomic_int_get(&initialized) && !g_atomic_int_get(&stopping)) {
		msg = g_async_queue_pop(messages);
		if(msg == &exit_message)
			break;

		//wait recoder fclose complete
		usleep(100000);

		JANUS_LOG(LOG_INFO, "process task file prefix_name:%s dir:%s codec:%s type:%d!\n", 
							msg->prefix_name,
							msg->dir, 
							msg->codec,
							msg->type);
		if(g_hash_table_contains(tuple_hash[msg->type], msg))
		{
			janus_processer_message *tuple_msg;
			JANUS_LOG(LOG_INFO, "process file %s!\n", msg->prefix_name);

			tuple_msg = g_hash_table_lookup(tuple_hash[msg->type], msg); 

			if(msg->type == JANUS_RECORDER_AUDIO)
			{
				run_script_async(msg->dir, msg->prefix_name, 
					msg->codec, tuple_msg->codec, 
					tuple_msg->business_model, tuple_msg->business_id);
			}
			else
			{
				run_script_async(msg->dir, msg->prefix_name, 
					tuple_msg->codec, msg->codec,
					tuple_msg->business_model, tuple_msg->business_id);
			}

			g_hash_table_remove(tuple_hash[msg->type], msg);
		}
		else
		{	
			g_hash_table_insert(pattern_hash[msg->type], msg, msg);
		}
	}

	JANUS_LOG(LOG_VERB, "Leaving Processer thread\n");
	return NULL;
}

